/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;

import com.alibaba.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import com.alibaba.rocketmq.client.impl.FindBrokerResult;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.client.log.ClientLogger;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.body.LockBatchRequestBody;
import com.alibaba.rocketmq.common.protocol.body.UnlockBatchRequestBody;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumeType;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;

/**
 * Base class for rebalance algorithm
 *
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-22
 */
public abstract class RebalanceImpl {
	protected static final Logger log = ClientLogger.getLog();
	protected final ConcurrentHashMap<MessageQueue, ProcessQueue> processQueueTable = new ConcurrentHashMap<MessageQueue, ProcessQueue>(64);
	protected final ConcurrentHashMap<String/* topic */, Set<MessageQueue>> topicSubscribeInfoTable = new ConcurrentHashMap<String, Set<MessageQueue>>();
	protected final ConcurrentHashMap<String /* topic */, SubscriptionData> subscriptionInner = new ConcurrentHashMap<String, SubscriptionData>();
	protected String consumerGroup;
	protected MessageModel messageModel;
	protected AllocateMessageQueueStrategy allocateMessageQueueStrategy;
	protected MQClientInstance mQClientFactory;

	public RebalanceImpl(String consumerGroup, MessageModel messageModel, AllocateMessageQueueStrategy allocateMessageQueueStrategy, MQClientInstance mQClientFactory) {
		this.consumerGroup = consumerGroup;
		this.messageModel = messageModel;
		this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
		this.mQClientFactory = mQClientFactory;
	}

	public abstract ConsumeType consumeType();

	public void unlock(final MessageQueue mq, final boolean oneway) {
		FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
		if (findBrokerResult != null) {
			UnlockBatchRequestBody requestBody = new UnlockBatchRequestBody();
			requestBody.setConsumerGroup(this.consumerGroup);
			requestBody.setClientId(this.mQClientFactory.getClientId());
			requestBody.getMqSet().add(mq);

			try {
				// 向该MessageQueue对象的BrokerName下面的主用Broker发送UNLOCK_BATCH_MQ请求码请求对MessageQueue进行解锁操作，
				// 采用oneway形式发送请求消息，发送成功则直接返回true，最后释放此互斥锁
				this.mQClientFactory.getMQClientAPIImpl().unlockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000, oneway);
				log.warn("unlock messageQueue. group:{}, clientId:{}, mq:{}", //
						this.consumerGroup, //
						this.mQClientFactory.getClientId(), //
						mq);
			} catch (Exception e) {
				log.error("unlockBatchMQ exception, " + mq, e);
			}
		}
	}

	public void unlockAll(final boolean oneway) {
		HashMap<String, Set<MessageQueue>> brokerMqs = this.buildProcessQueueTableByBrokerName();

		for (final Map.Entry<String, Set<MessageQueue>> entry : brokerMqs.entrySet()) {
			final String brokerName = entry.getKey();
			final Set<MessageQueue> mqs = entry.getValue();

			if (mqs.isEmpty())
				continue;

			FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true);
			if (findBrokerResult != null) {
				UnlockBatchRequestBody requestBody = new UnlockBatchRequestBody();
				requestBody.setConsumerGroup(this.consumerGroup);
				requestBody.setClientId(this.mQClientFactory.getClientId());
				requestBody.setMqSet(mqs);

				try {
					this.mQClientFactory.getMQClientAPIImpl().unlockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000, oneway);

					for (MessageQueue mq : mqs) {
						ProcessQueue processQueue = this.processQueueTable.get(mq);
						if (processQueue != null) {
							processQueue.setLocked(false);
							log.info("the message queue unlock OK, Group: {} {}", this.consumerGroup, mq);
						}
					}
				} catch (Exception e) {
					log.error("unlockBatchMQ exception, " + mqs, e);
				}
			}
		}
	}

	private HashMap<String/* brokerName */, Set<MessageQueue>> buildProcessQueueTableByBrokerName() {
		HashMap<String, Set<MessageQueue>> result = new HashMap<String, Set<MessageQueue>>();
		for (MessageQueue mq : this.processQueueTable.keySet()) { // ConcurrentHashMap <MessageQueue,ProcessQueue>
			Set<MessageQueue> mqs = result.get(mq.getBrokerName());
			if (null == mqs) {
				mqs = new HashSet<MessageQueue>();
				// 从RebalanceImpl.processQueueTable:ConcurrentHashMap <MessageQueue,ProcessQueue>变量中根据所有的Key值构建一个结构是HashMap<String/* brokerName */, Set<MessageQueue>>的Map集合
				result.put(mq.getBrokerName(), mqs);
			}
			// 按每个brokerName来构建MessageQueue集合
			mqs.add(mq);
		}

		return result;
	}

	public boolean lock(final MessageQueue mq) {
		FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
		if (findBrokerResult != null) {
			LockBatchRequestBody requestBody = new LockBatchRequestBody();
			requestBody.setConsumerGroup(this.consumerGroup);
			requestBody.setClientId(this.mQClientFactory.getClientId());
			requestBody.getMqSet().add(mq);

			try {
				Set<MessageQueue> lockedMq = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
				for (MessageQueue mmqq : lockedMq) {
					ProcessQueue processQueue = this.processQueueTable.get(mmqq);
					if (processQueue != null) {
						processQueue.setLocked(true);
						processQueue.setLastLockTimestamp(System.currentTimeMillis());
					}
				}

				boolean lockOK = lockedMq.contains(mq);
				log.info("the message queue lock {}, {} {}", //
						(lockOK ? "OK" : "Failed"), //
						this.consumerGroup, //
						mq);
				return lockOK;
			} catch (Exception e) {
				log.error("lockBatchMQ exception, " + mq, e);
			}
		}

		return false;
	}

	public void lockAll() {
		// 12.1、按每个brokerName来构建MessageQueue集合
		HashMap<String/* brokerName */, Set<MessageQueue>> brokerMqs = this.buildProcessQueueTableByBrokerName();

		Iterator<Entry<String, Set<MessageQueue>>> it = brokerMqs.entrySet().iterator();
		// 遍历该Map集合的每个brokerName
		while (it.hasNext()) {
			Entry<String, Set<MessageQueue>> entry = it.next();
			final String brokerName = entry.getKey();
			final Set<MessageQueue> mqs = entry.getValue();

			if (mqs.isEmpty())
				// 12.2）若brokerName对应的MessageQueue集合为空则继续遍历下一个brokerName，直到遍历完为止
				continue;

			// 若对应的MessageQueue集合不为空，则继续下面的操作；
			// 12.3)以brokerName为参数从MQClientInstance.brokerAddrTable中获取主用Broker的地址；若主用Broker地址为空则继续遍历下一个brokerName，直到遍历完为止；否则继续下面的操作
			FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(brokerName, MixAll.MASTER_ID, true);
			// 主用Broker地址不为空
			if (findBrokerResult != null) {
				// 12.4）初始化LockBatchRequestBody对象，其中MqSet等于该brokerName对应的MessageQueue集合、ClientId等于该Consumer的ClientID；
				LockBatchRequestBody requestBody = new LockBatchRequestBody();
				requestBody.setConsumerGroup(this.consumerGroup);
				requestBody.setClientId(this.mQClientFactory.getClientId());
				requestBody.setMqSet(mqs);

				try {
					// 12.5）向该brokerName下面的主用Broker发送LOCK_BATCH_MQ请求码的请求消息，请求Broker将发送的MessageQueue集合加锁
					// LOCK_BATCH_MQ表示Broker Consumer向Master锁定队列
					// 返回已经锁住的MessageQueue集合lockOKMQSet
					Set<MessageQueue> lockOKMQSet = this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);

					// 12.6）根据Broker返回的已经锁住的MessageQueue集合,将processQueueTable中已经锁住的MessageQueue对象对应的ProcessQueue对象的locked置为true、更新lastLockTimestamp值
					for (MessageQueue mq : lockOKMQSet) {
						ProcessQueue processQueue = this.processQueueTable.get(mq); // ConcurrentHashMap<MessageQueue, ProcessQueue>
						if (processQueue != null) {
							if (!processQueue.isLocked()) {
								log.info("the message queue locked OK, Group: {} {}", this.consumerGroup, mq);
							}

							processQueue.setLocked(true);
							processQueue.setLastLockTimestamp(System.currentTimeMillis());
						}
					}
					// 将未锁住的MessageQueue对应的ProcessQueue对象的locked置为false；
					for (MessageQueue mq : mqs) {
						if (!lockOKMQSet.contains(mq)) {
							ProcessQueue processQueue = this.processQueueTable.get(mq);
							if (processQueue != null) {
								processQueue.setLocked(false);
								log.warn("the message queue locked Failed, Group: {} {}", this.consumerGroup, mq);
							}
						}
					}
				} catch (Exception e) {
					log.error("lockBatchMQ exception, " + mqs, e);
				}
			}
		}
	}

	public void doRebalance() {
		Map<String/* topic */, SubscriptionData> subTable = this.getSubscriptionInner();
		if (subTable != null) {
			// 消费端遍历自己的所有topic，依次调rebalanceByTopic
			for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
				final String topic = entry.getKey();
				try {
					// *****************************************
					// 为topic所分配的MessageQueue创建ProcessQueue对象，并且创建拉取消息请求放入后台服务线程中自动执行拉取逻辑；
					this.rebalanceByTopic(topic);
				} catch (Exception e) {
					if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
						log.warn("rebalanceByTopic Exception", e);
					}
				}
			}
		}

		// 遍历RebalanceImpl.processQueueTable队列中的每个MessageQueue对象的topic，
		// 若该topic不在RebalanceImpl.subscriptionInner: ConcurrentHashMap<String /* topic */, SubscriptionData>列表中，
		// 则从RebalanceImpl.processQueueTable队列将该MessageQueue对象的记录删掉，然后置对应的ProcessQueue对象的dropped为true；
		this.truncateMessageQueueNotMyTopic();
	}

	private void rebalanceByTopic(final String topic) {
		switch (messageModel) {
			case BROADCASTING : {
				// 根据topic获取此topic下的所有queue
				// 1、以topic值为key值从RebalanceImpl.topicSubscribeInfoTable: ConcurrentHashMap<String/* topic */, Set<MessageQueue>>列表中获取MessageQueue集合；
				Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);
				if (mqSet != null) {
					// 1、为集合中的每个MessageQueue对象创建ProcessQueue对象，
					// 2、对于PUSH模式将拉取消息的请求放入服务线程（PullMessageService）中。
					boolean changed = this.updateProcessQueueTableInRebalance(topic, mqSet);
					if (changed) {
						// 对于PUSH模式下面没有相应的处理逻辑，即messageQueueChanged方法为空；
						// 对于PULL模式下会调用RebalancePullImpl对象的messageQueueChanged方法
						// 此方法的目的是对于采用了计划消息拉取服务的应用来说，当Consumer订阅的MessageQueue有变动或者ProcessQueue有更新时触发消息拉取动作
						this.messageQueueChanged(topic, mqSet, mqSet);
						log.info("messageQueueChanged {} {} {} {}", //
								consumerGroup, //
								topic, //
								mqSet, //
								mqSet);
					}
				} else {
					log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
				}
				break;
			}
			case CLUSTERING : {
				// 根据topic获取此topic下的所有queue
				// 1、以入参topic值从RebalanceImpl.topicSubscribeInfoTable: ConcurrentHashMap<String/* topic */, Set<MessageQueue>>变量中获取对应的MessageQueue集合（命名为mqSet）；
				Set<MessageQueue> mqSet = this.topicSubscribeInfoTable.get(topic);

				// 选择一台broker获取基于group的所有消费端（有心跳向所有broker注册客户端信息）
				List<String> cidAll = this.mQClientFactory.findConsumerIdList(topic, consumerGroup);
				if (null == mqSet) {
					if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
						log.warn("doRebalance, {}, but the topic[{}] not exist.", consumerGroup, topic);
					}
				}

				if (null == cidAll) {
					log.warn("doRebalance, {} {}, get consumer id list failed", consumerGroup, topic);
				}

				/**
				 * 5、若第1步的MessageQueue集合（名为mqSet）和上一步的ClientId集合都不为null则继续进行下面操作，否则直接退出该方法；
				 */
				if (mqSet != null && cidAll != null) {
					List<MessageQueue> mqAll = new ArrayList<MessageQueue>();
					mqAll.addAll(mqSet);
					// 将所有queue排好序类似于记录
					Collections.sort(mqAll);
					// 将所有消费端consumer排好序，相当于页数
					Collections.sort(cidAll);
					// 选择队列分配策略实例AllocateMessageQueueStrategy执行分配算法
					// 平均分配算法，其实是类似于分页的算法
					AllocateMessageQueueStrategy strategy = this.allocateMessageQueueStrategy;

					List<MessageQueue> allocateResult = null;
					try {
						// 获取当前consumer所在页面应该分配到的queue
						/**
						 * 6、调用RebalanceImpl. allocateMessageQueueStrategy.allocate (String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll)
						 *    为当前Consumer分配应该消费的MessageQueue集合队列（命名allocateResultSet）；
						 *    有四种分配策略，包括平均分配（AVG）、AVG_BY_CIRCLE 、CONFIG 、MACHINE_ROOM，默认为AllocateMessageQueueAveragely分配策略；
						 */
						allocateResult = strategy.allocate(//
								this.consumerGroup, //
								this.mQClientFactory.getClientId(), //
								mqAll, //
								cidAll);
					} catch (Throwable e) {
						log.error("AllocateMessageQueueStrategy.allocate Exception. allocateMessageQueueStrategyName={}", strategy.getName(), e);
						return;
					}

					Set<MessageQueue> allocateResultSet = new HashSet<MessageQueue>();
					if (allocateResult != null) {
						allocateResultSet.addAll(allocateResult);
					}

					// 一个MessageQueue消息队列的消息分配一个消息处理队列ProcessQueue来进行消费处理，
					// 这两个队列的关系存储在RebalanceImpl.processQueueTable队列变量中。
					/**
					 * *****************************************************************************
					 * 6、以topic和上一步分配的MessageQueue集合（名为allocateResultSet）为参数
					 * 调用RebalanceImpl. updateProcessQueueTableInRebalance (String topic, Set<MessageQueue> mqSet)方法，
					 * 1、对每个MessageQueue创建一个ProcessQueue对象并存入RebalanceImpl.processQueueTable队列中，并将processQueueTable队列中其他的MessageQueue记录删除掉；
					 * 2、对于PUSH模式要创建PullRequest对象并放入消息拉取线程（PullMessageService）的pullRequestQueue队列中，
					 * 最后一个标记位表示是否对RebalanceImpl.processQueueTable列表有修改，若有则返回true；
					 */
					boolean changed = this.updateProcessQueueTableInRebalance(topic, allocateResultSet);
					if (changed) {
						log.info("rebalanced allocate source. allocateMessageQueueStrategyName={}, group={}, topic={}, mqAllSize={}, cidAllSize={}, mqAll={}, cidAll={}", strategy.getName(), consumerGroup, topic, mqSet.size(), cidAll.size(), mqSet, cidAll);
						log.info("rebalanced result changed. allocateMessageQueueStrategyName={}, group={}, topic={}, ConsumerId={}, rebalanceSize={}, rebalanceMqSet={}", strategy.getName(), consumerGroup, topic, this.mQClientFactory.getClientId(), allocateResultSet.size(), mqAll.size(), cidAll.size(), allocateResultSet);

						/**
						 * 7、若上一步的返回结果为true（即RebalanceImpl.processQueueTable列表有记录变化时），
						 * 则调用RebalanceImpl.messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided)方法，
						 * 其中mqAll为第1步的MessageQueue集合（名为mqSet），mqDivided为第6步的MessageQueue集合（名为allocateResultSet）。
						 * 对于PUSH模式下面没有相应的处理逻辑，即messageQueueChanged方法为空；
						 * 对于PULL模式下会调用RebalancePullImpl对象的messageQueueChanged方法，
						 * 此方法的目的是对于采用了计划消息拉取服务的应用来说，当Consumer订阅的MessageQueue有变动或者或者ProcessQueue有更新时触发消息拉取动作
						 */
						this.messageQueueChanged(topic, mqSet, allocateResultSet);
					}
				}
				break;
			}
			default :
				break;
		}
	}

	public abstract void messageQueueChanged(final String topic, final Set<MessageQueue> mqAll, final Set<MessageQueue> mqDivided);

	public void removeProcessQueue(final MessageQueue mq) {
		ProcessQueue prev = this.processQueueTable.remove(mq);
		if (prev != null) {
			boolean droped = prev.isDropped();
			prev.setDropped(true);
			this.removeUnnecessaryMessageQueue(mq, prev);
			log.info("Fix Offset, {}, remove unnecessary mq, {} Droped: {}", consumerGroup, mq, droped);
		}
	}

	private boolean updateProcessQueueTableInRebalance(final String topic, final Set<MessageQueue> mqSet) {
		boolean changed = false;

		Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<MessageQueue, ProcessQueue> next = it.next();
			MessageQueue mq = next.getKey();
			ProcessQueue pq = next.getValue();

			// 1、遍历RebalanceImpl.processQueueTable:ConcurrentHashMap <MessageQueue, ProcessQueue>集合，检查该集合中的每条记录，
			// 对于不在入参MessageQueue集合中记录
			// 或者记录中ProcessQueue对象的距离上次拉取时间已经超时，
			// 则从该列表中删除掉该记录：

			// 1.1）首先检查MessageQueue的topic值是否等于参数topic；若不相等，则继续遍历下一条记录；若相等，则继续下面的处理逻辑；
			if (mq.getTopic().equals(topic)) {
				// 1.2) 检查满足第1步的MessageQueue对象是否在入参MessageQueue集合中
				if (!mqSet.contains(mq)) {
					// 若不在此集合中，则将该记录对应的ProcessQueue对象的dropped变量置为true，
					// 并且调用 removeUnnecessaryMessageQueue 方法删除该消息队列的消费进度（详见4小节）
					pq.setDropped(true);

					// **************删除未使用的消息队列的消费进度****************
					// 该方法主用目的是：首先将指定MessageQueue消息队列的消费进度向Broker同步；然后从本地删除该消息队列的消费进度；最后对于PUSH模式下面的顺序消费从Broker端将消息队列解锁。
					// push
					// pull 大致逻辑与PUSH消费模式的第1/2步逻辑一样，只是没有释放分布式锁的处理逻辑
					if (this.removeUnnecessaryMessageQueue(mq, pq)) {
						// 删除成功则再将该消息队列的记录从RebalanceImpl.processQueueTable列表中删除并置 changed=true
						it.remove();
						changed = true;
						log.info("doRebalance, {}, remove unnecessary mq, {}", consumerGroup, mq);
					}
					// 若在此集合中，则检查ProcessQueue距离上次拉取时间（ProcessQueue. lastPullTimestamp）是否已经超时（默认120秒）
				} else if (pq.isPullExpired()) {
					switch (this.consumeType()) {
						case CONSUME_ACTIVELY :
							break;
						case CONSUME_PASSIVELY :
							// 若已经超时，并且消费类型为被动消费（即为PUSH模式，RebalancePushImpl），
							// 则将该记录对应的ProcessQueue对象的dropped变量置为true，并且调用 removeUnnecessaryMessageQueue 方法删除该消息队列的消费进度（详见4小节）
							pq.setDropped(true);

							// **************删除未使用的消息队列的消费进度****************
							// 该方法主用目的是：首先将指定MessageQueue消息队列的消费进度向Broker同步；然后从本地删除该消息队列的消费进度；最后对于PUSH模式下面的顺序消费从Broker端将消息队列解锁。
							// push
							// pull
							if (this.removeUnnecessaryMessageQueue(mq, pq)) {
								// 若删除成功则再将遍历到的此条记录从RebalanceImpl.processQueueTable列表中删除并置 changed=true ；
								it.remove();
								changed = true;
								log.error("[BUG]doRebalance, {}, remove unnecessary mq, {}, because pull is pause, so try to fixed it", consumerGroup, mq);
							}
							break;
						default :
							break;
					}
				}
			}
		}

		/**
		 * 2、遍历入参MessageQueue集合，检查该集合中的每个MessageQueue对象，
		 * 保证每个MessageQueue对象在processQueueTable队列中都有ProcessQueue对象。
		 * 对于不在RebalanceImpl.processQueueTable:ConcurrentHashMap <MessageQueue, ProcessQueue>列表中的MessageQueue对象，
		 * 要构建PullRequest对象以及processQueueTable集合的K-V值
		 */
		List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
		for (MessageQueue mq : mqSet) {
			if (!this.processQueueTable.containsKey(mq)) {
				// 对于不在RebalanceImpl.processQueueTable:ConcurrentHashMap <MessageQueue, ProcessQueue>列表中的MessageQueue对象，
				// 2.1）初始化PullRequest对象，并设置consumerGroup、messageQueue变量值， 初始化新的ProcessQueue对象 赋值给PullRequest对象的processQueue变量；
				PullRequest pullRequest = new PullRequest();
				pullRequest.setConsumerGroup(consumerGroup);
				pullRequest.setMessageQueue(mq);
				pullRequest.setProcessQueue(new ProcessQueue());

				// *********************获取MessageQueue队列的下一个消费偏移量********************************************
				// 2.2）调用RebalanceImpl. computePullFromWhere (MessageQueue mq)方法获取该MessageQueue对象的下一个消费偏移值offset
				// pull 直接返回0
				// push
				long nextOffset = this.computePullFromWhere(mq);
				if (nextOffset >= 0) {
					/**
					 * 2.3）若该offset值大于等于0，
					 * 则首先用该offset值设置PullRequest对象的nextOffset值；
					 * 然后将该PullRequest对象放入局部变量pullRequestList:List<PullRequest>集合中；
					 * 再将遍历到的MessageQueue对象和该PullRequest对象的processQueue变量存入RebalanceImpl.processQueueTable变量中；
					 * 最后置 change等于true ；
					 */
					pullRequest.setNextOffset(nextOffset);
					pullRequestList.add(pullRequest);
					changed = true;
					this.processQueueTable.put(mq, pullRequest.getProcessQueue());
					log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
				} else {
					log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
				}
			}
		}

		// 对于PUSH模式将拉取消息的请求放入服务线程（PullMessageService）中。
		/**
		 * 3、以上一步的局部变量pullRequestList集合为参数调用RebalanceImpl.dispatchPullRequest(List<PullRequest> pullRequestList)方法。
		 * 该方法的目的是将拉取消息的请求对象PullRequest放入拉取消息的服务线程（PullMessageService）中，
		 * 并在后台执行拉取消息的业务逻辑。
		 * 对于PULL模式，是由应用层触发拉取消息，
		 * 故在RebalancePullImpl子类中该方法没有业务逻辑
		 */
		this.dispatchPullRequest(pullRequestList);

		// 4、返回 change 的值，该值在进入updateProcessQueueTableInRebalance方法时初始化为false，若在上述步骤中没有重置则返回false；
		return changed;
	}

	public abstract boolean removeUnnecessaryMessageQueue(final MessageQueue mq, final ProcessQueue pq);

	public abstract void dispatchPullRequest(final List<PullRequest> pullRequestList);

	public abstract long computePullFromWhere(final MessageQueue mq);

	private void truncateMessageQueueNotMyTopic() {
		Map<String/* topic */, SubscriptionData> subTable = this.getSubscriptionInner();

		for (MessageQueue mq : this.processQueueTable.keySet()) {
			if (!subTable.containsKey(mq.getTopic())) {
				ProcessQueue pq = this.processQueueTable.remove(mq);
				if (pq != null) {
					pq.setDropped(true);
					log.info("doRebalance, {}, truncateMessageQueueNotMyTopic remove unnecessary mq, {}", consumerGroup, mq);
				}
			}
		}
	}

	public ConcurrentHashMap<String, SubscriptionData> getSubscriptionInner() {
		return subscriptionInner;
	}

	public ConcurrentHashMap<MessageQueue, ProcessQueue> getProcessQueueTable() {
		return processQueueTable;
	}

	public ConcurrentHashMap<String, Set<MessageQueue>> getTopicSubscribeInfoTable() {
		return topicSubscribeInfoTable;
	}

	public String getConsumerGroup() {
		return consumerGroup;
	}

	public void setConsumerGroup(String consumerGroup) {
		this.consumerGroup = consumerGroup;
	}

	public MessageModel getMessageModel() {
		return messageModel;
	}

	public void setMessageModel(MessageModel messageModel) {
		this.messageModel = messageModel;
	}

	public AllocateMessageQueueStrategy getAllocateMessageQueueStrategy() {
		return allocateMessageQueueStrategy;
	}

	public void setAllocateMessageQueueStrategy(AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
		this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
	}

	public MQClientInstance getmQClientFactory() {
		return mQClientFactory;
	}

	public void setmQClientFactory(MQClientInstance mQClientFactory) {
		this.mQClientFactory = mQClientFactory;
	}

	public void destroy() {
		Iterator<Entry<MessageQueue, ProcessQueue>> it = this.processQueueTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<MessageQueue, ProcessQueue> next = it.next();
			next.getValue().setDropped(true);
		}

		this.processQueueTable.clear();
	}
}
